package com.spzx.common.rabbit.config;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbit.entity.GuiguCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallback();
    }

    /**
     * 初始化确认回调和回退回调两个函数
     */
    private void setupCallback() {

        /**
         * 交换机是否接收到消息，都会回调confirm函数
         */
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                if(ack){ //交换机收到消息了
                    log.info("confirm - correlationData="+correlationData);
                }else{ //交换机没收到消息
                    log.error("confirm - cause="+cause);

                    //消息发送失败，重发消息
                    retrySendMsg(correlationData);
                }
            }
        });

        /**
         * 交换机转发队列失败会执行回退函数。交换机转发消息到队列成功了，那么不执行回退函数。
         */
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                /*try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/


//                returned=ReturnedMessage [message=(Body:'hello, confirm' MessageProperties
//                    [headers={spring_returned_message_correlation=mq:762ed3fa53fa48fa94045c7b55817bd0},
//                contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT,
//                        priority=0, deliveryTag=0]), replyCode=312, replyText=NO_ROUTE, exchange=spzx.test, routingKey=xxx]
                log.error("returned="+returned);

                //发送消息时携带关联数据的id
                String uuid = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
                String guiguCorrelationDataStr = (String)redisTemplate.opsForValue().get(uuid);
                GuiguCorrelationData guiguCorrelationData = JSON.parseObject(guiguCorrelationDataStr, GuiguCorrelationData.class);

                //问题：延迟插件发送消息到交换机是成功的，但是由于延迟转发给队列。会导致，执行回退函数进行消息重复。
                //解决方案1：如果是延迟消息，那么，就不进行重试发送
                if(guiguCorrelationData.isDelay()){
                    log.info("********* 延迟插件发送延迟消息被回退，不再重发消息了 *********");
                    return;
                }


                //消息发送失败，重发消息
                retrySendMsg(guiguCorrelationData);
            }
        });

    }


    public void retrySendMsg(CorrelationData correlationData){
        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData)correlationData;
        int retryCount = guiguCorrelationData.getRetryCount();
        if(retryCount>=3){
            log.info("发送消息重试次数已经达到3次！"); //可以发短信或邮件通知管理员；
            return;
        }

        retryCount ++;
        guiguCorrelationData.setRetryCount(retryCount);
        //更新缓存重试次数
        redisTemplate.opsForValue().set(guiguCorrelationData.getId(), JSON.toJSONString(guiguCorrelationData),10, TimeUnit.MINUTES);
        log.info("进行消息重发！");

        if(guiguCorrelationData.isDelay()){ //延迟消息重新发送
            log.info("延迟消息又重发了....");
            //方案2：延迟消息也进行再次重发
            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(), guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(), new MessagePostProcessor() {
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            message.getMessageProperties().setDelay(guiguCorrelationData.getDelayTime()*1000); //单位毫秒
                            return message;
                        }
                    }, guiguCorrelationData);

        }else{
            //不是延迟消息重发消息
            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(),guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),guiguCorrelationData);
        }

    }
}
